package com.wydstation.power.collectionserver.consumer;


import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * <p>
 *
 * @author <714037465@qq.com>
 * @since 2018/8/2023:18
 **/
@Configuration
public class KafkaDataConsumer {
    private static final String KAFKA_INFLUXDB_TOPIC = "modbus-2";

    private Properties init() {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        properties.setProperty("group.id", "send_demo_group");
        properties.setProperty("client.id", "zhang_test1");
        properties.setProperty("auto.offset.reset", "earliest");
        properties.setProperty("enable.auto.commit", "true");
        return properties;
    }

    public void run() {
        KafkaConsumer<String,byte[]> consumer = new KafkaConsumer(init());
        consumer.subscribe(Collections.singletonList(KAFKA_INFLUXDB_TOPIC));
        for (; ; ) {
            //不要在这处理逻辑
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(5));
            System.out.println("-------------------->"+records.isEmpty());
            records.forEach(recode -> {
                System.out.println(recode.key() + " " + recode.offset() + "  " + recode.value() + "-->" + recode.partition());
            });
            consumer.commitSync();
        }
    }

    public static void main(String[] args) {
        KafkaDataConsumer consumer=new KafkaDataConsumer();
        consumer.run();
    }
}
